/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.TreeSet;

import org.slf4j.Logger;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.datatype.DCPackableString;
import org.sss.mapreduce.datatype.PackableInt;
import org.sss.mapreduce.datatype.PackableString;
import org.sss.server.SssServer;

public class ContTopXSelectReducer extends Reducer {
  private final Logger logger = SssServer.getLogger(ContTopXSelectReducer.class);
  private final Buffer buffer = new Buffer();
  private int rank = 10;
  private int scheduled = 0;
  @Override
  public void configure(Context context) throws Exception {
    Configuration conf = context.getConfiguration();
    this.rank = conf.getInt("topx.select.rank", 10);
    this.scheduled = 0;
  }
  public void reduce(Context context,
                     DCPackableString key, Iterable<PackableInt> values,
                     Output<PackableString, PackableInt> output)
    throws Exception {
    logger.trace("reduce method");
    int sum = 0;
    for (PackableInt v: values) {
      sum += v.get();
    }
    buffer.addEntry(key.get(), sum);
  }
  public void cleanup(Context context,
                      Output<PackableString, PackableInt> output)
    throws SssException {
    logger.trace("cleanup {}", scheduled);
    buffer.write(output, rank);
    scheduled++;
  }
  private static class Buffer {
    private final Logger logger = SssServer.getLogger(Buffer.class);
    private Map<String, Integer> buffer =
      Collections.synchronizedMap(new HashMap<String, Integer>());
    public void addEntry(String key, int value) {
      buffer.put(key, new Integer(value));
    }
    private int getLowerBound(int rank) {
      TreeSet<Integer> counts = new TreeSet<Integer>(buffer.values());
      if (counts.size() <= rank) {
        return 0;
      }
      else {
        Iterator<Integer> it = counts.descendingIterator();
        Integer count = null;
        for (int i=0; i<rank; i++) {
          count = it.next();
          logger.debug("top counts - {}", count.intValue());
        }
        return count.intValue();
      }
    }
    public void write(Output<PackableString, PackableInt> output, int rank)
      throws SssException {
      int lowerBound = getLowerBound(rank);
      for (Map.Entry<String, Integer> e: buffer.entrySet()) {
        if (e.getValue().intValue() >= lowerBound) {
          output.write(new PackableString(e.getKey()), new PackableInt(e.getValue()));
        }
      }
      buffer.clear();
    }
  }
}
